/*
    SUSAN® - Sole of Unix Save ANything

   Copyright (C) 2001-2012 Hechzh Property
   Copyright (C) 2016-2020 Hechzh Property

   
*/
/*
 * Kern Sibbald, April MMI
 * added BB02 format October MMII
 */
/**
 * @file
 * Volume (tape/disk) record handling functions
 */

#include "include/susan.h"
#include "stored/jcr_private.h"
#include "stored/stored.h"
#include "stored/device_control_record.h"
#include "lib/attribs.h"
#include "lib/util.h"
#include "include/jcr.h"

namespace storagedaemon {

/**
 * Convert a FileIndex into a printable
 * ASCII string.  Not reentrant.
 *
 * If the FileIndex is negative, it flags the
 * record as a Label, otherwise it is simply
 * the FileIndex of the current file.
 */
const char* FI_to_ascii(char* buf, int fi)
{
  if (fi >= 0) {
    sprintf(buf, "%d", fi);
    return buf;
  }
  switch (fi) {
    case PRE_LABEL:
      return "PRE_LABEL";
    case VOL_LABEL:
      return "VOL_LABEL";
    case EOM_LABEL:
      return "EOM_LABEL";
    case SOS_LABEL:
      return "SOS_LABEL";
    case EOS_LABEL:
      return "EOS_LABEL";
    case EOT_LABEL:
      return "EOT_LABEL";
      break;
    case SOB_LABEL:
      return "SOB_LABEL";
      break;
    case EOB_LABEL:
      return "EOB_LABEL";
      break;
    default:
      sprintf(buf, _("unknown: %d"), fi);
      return buf;
  }
}


static const char* compression_to_str(PoolMem& resultbuffer,
                                      const char* compression_algorithm,
                                      uint32_t data_length,
                                      uint16_t compression_level,
                                      uint16_t compression_algorithm_version)
{
  PoolMem tmp(PM_MESSAGE);
  tmp.bsprintf("%s, level=%u, version=%u, length=%u", compression_algorithm,
               compression_level, compression_algorithm_version, data_length);
  resultbuffer.strcat(tmp);
  return resultbuffer.c_str();
}

static const char* record_compression_to_str(PoolMem& resultbuffer,
                                             const DeviceRecord* rec)
{
  int32_t maskedStream = rec->maskedStream;
  POOLMEM* buf = rec->data;
  PoolMem tmp(PM_MESSAGE);
  unser_declare;

  if (maskedStream == STREAM_SPARSE_GZIP_DATA ||
      maskedStream == STREAM_SPARSE_COMPRESSED_DATA) {
    uint64_t faddr = 0;

    SerBegin(buf, sizeof(uint64_t));
    unser_uint64(faddr);
    SerEnd(buf, sizeof(uint64_t));

    buf += sizeof(uint64_t);

    Dmsg1(400, "Sparse data stream found: start address=%llu\n", faddr);
    tmp.bsprintf("Sparse: StartAddress=%llu. ", faddr);
    resultbuffer.strcat(tmp);
  }

  Dmsg1(400, "Stream found in DecompressData(): %d\n", maskedStream);
  switch (maskedStream) {
    case STREAM_COMPRESSED_DATA:
    case STREAM_SPARSE_COMPRESSED_DATA:
    case STREAM_WIN32_COMPRESSED_DATA:
    case STREAM_ENCRYPTED_FILE_COMPRESSED_DATA:
    case STREAM_ENCRYPTED_WIN32_COMPRESSED_DATA: {
      uint32_t comp_magic, comp_len;
      uint16_t comp_level, comp_version;

      /*
       * Read compress header
       */
      UnserBegin(buf, sizeof(comp_stream_header));
      unser_uint32(comp_magic);
      unser_uint32(comp_len);
      unser_uint16(comp_level);
      unser_uint16(comp_version);
      UnserEnd(buf, sizeof(comp_stream_header));

      Dmsg4(400,
            "Compressed data stream found: magic=0x%x, len=%d, level=%d, "
            "ver=0x%x\n",
            comp_magic, comp_len, comp_level, comp_version);

      switch (comp_magic) {
        case COMPRESS_GZIP:
          compression_to_str(resultbuffer, "GZIP", comp_len, comp_level,
                             comp_version);
          break;
        case COMPRESS_LZO1X:
          compression_to_str(resultbuffer, "LZO1X", comp_len, comp_level,
                             comp_version);
          break;
        case COMPRESS_FZFZ:
          compression_to_str(resultbuffer, "FZFZ", comp_len, comp_level,
                             comp_version);
          break;
        case COMPRESS_FZ4L:
          compression_to_str(resultbuffer, "FZ4L", comp_len, comp_level,
                             comp_version);
          break;
        case COMPRESS_FZ4H:
          compression_to_str(resultbuffer, "FZ4H", comp_len, comp_level,
                             comp_version);
          break;
        default:
          tmp.bsprintf(
              _("Compression algorithm 0x%x found, but not supported!\n"),
              comp_magic);
          resultbuffer.strcat(tmp);
          break;
      }
      break;
    }
    case STREAM_GZIP_DATA:
    case STREAM_SPARSE_GZIP_DATA:
      /* deprecated */
      compression_to_str(resultbuffer, "GZIP", 0, 0, 0);
      break;
    default:
      break;
  }

  return resultbuffer.c_str();
}

static const char* record_digest_to_str(PoolMem& resultbuffer,
                                        const DeviceRecord* rec)
{
  char digest[BASE64_SIZE(CRYPTO_DIGEST_MAX_SIZE)];

  switch (rec->maskedStream) {
    case STREAM_MD5_DIGEST:
      BinToBase64(digest, sizeof(digest), (char*)rec->data,
                  CRYPTO_DIGEST_MD5_SIZE, true);
      break;
    case STREAM_SHA1_DIGEST:
      BinToBase64(digest, sizeof(digest), (char*)rec->data,
                  CRYPTO_DIGEST_SHA1_SIZE, true);
      break;
    case STREAM_SHA256_DIGEST:
      BinToBase64(digest, sizeof(digest), (char*)rec->data,
                  CRYPTO_DIGEST_SHA256_SIZE, true);
      break;
    case STREAM_SHA512_DIGEST:
      BinToBase64(digest, sizeof(digest), (char*)rec->data,
                  CRYPTO_DIGEST_SHA512_SIZE, true);
      break;
    default:
      return "";
  }

  resultbuffer.bsprintf("%s (base64)", digest);

  return resultbuffer.c_str();
}

/**
 * Convert a Stream ID into a printable
 * ASCII string. Not reentrant.
 *
 * A negative stream number represents
 * stream data that is continued from a
 * record in the previous block.
 *
 * If the FileIndex is negative, we are
 * dealing with a Label, hence the
 * stream is the JobId.
 */
const char* stream_to_ascii(char* buf, int stream, int fi)
{
  if (fi < 0) {
    sprintf(buf, "%d", stream);
    return buf;
  }

  if (stream < 0) {
    stream = -stream;
    stream &= STREAMMASK_TYPE;
    /*
     * Stream was negative => all are continuation items
     */
    switch (stream) {
      case STREAM_UNIX_ATTRIBUTES:
        return "contUATTR";
      case STREAM_FILE_DATA:
        return "contDATA";
      case STREAM_WIN32_DATA:
        return "contWIN32-DATA";
      case STREAM_WIN32_GZIP_DATA:
        return "contWIN32-GZIP";
      case STREAM_WIN32_COMPRESSED_DATA:
        return "contWIN32-COMPRESSED";
      case STREAM_MD5_DIGEST:
        return "contMD5";
      case STREAM_SHA1_DIGEST:
        return "contSHA1";
      case STREAM_GZIP_DATA:
        return "contGZIP";
      case STREAM_COMPRESSED_DATA:
        return "contCOMPRESSED";
      case STREAM_UNIX_ATTRIBUTES_EX:
        return "contUNIX-Attributes-EX";
      case STREAM_RESTORE_OBJECT:
        return "contRESTORE-OBJECT";
      case STREAM_SPARSE_DATA:
        return "contSPARSE-DATA";
      case STREAM_SPARSE_GZIP_DATA:
        return "contSPARSE-GZIP";
      case STREAM_SPARSE_COMPRESSED_DATA:
        return "contSPARSE-COMPRESSED";
      case STREAM_PROGRAM_NAMES:
        return "contPROG-NAMES";
      case STREAM_PROGRAM_DATA:
        return "contPROG-DATA";
      case STREAM_MACOS_FORK_DATA:
        return "contMACOS-RSRC";
      case STREAM_HFSPLUS_ATTRIBUTES:
        return "contHFSPLUS-Attributes";
      case STREAM_SHA256_DIGEST:
        return "contSHA256";
      case STREAM_SHA512_DIGEST:
        return "contSHA512";
      case STREAM_SIGNED_DIGEST:
        return "contSIGNED-DIGEST";
      case STREAM_ENCRYPTED_SESSION_DATA:
        return "contENCRYPTED-SESSION-DATA";
      case STREAM_ENCRYPTED_FILE_DATA:
        return "contENCRYPTED-FILE";
      case STREAM_ENCRYPTED_FILE_GZIP_DATA:
        return "contENCRYPTED-GZIP";
      case STREAM_ENCRYPTED_FILE_COMPRESSED_DATA:
        return "contENCRYPTED-COMPRESSED";
      case STREAM_ENCRYPTED_WIN32_DATA:
        return "contENCRYPTED-WIN32-DATA";
      case STREAM_ENCRYPTED_WIN32_GZIP_DATA:
        return "contENCRYPTED-WIN32-GZIP";
      case STREAM_ENCRYPTED_WIN32_COMPRESSED_DATA:
        return "contENCRYPTED-WIN32-COMPRESSED";
      case STREAM_ENCRYPTED_MACOS_FORK_DATA:
        return "contENCRYPTED-MACOS-RSRC";
      case STREAM_PLUGIN_NAME:
        return "contPLUGIN-NAME";
      default:
        sprintf(buf, "%d", -stream);
        return buf;
    }
  }

  switch (stream & STREAMMASK_TYPE) {
    case STREAM_UNIX_ATTRIBUTES:
      return "UATTR";
    case STREAM_FILE_DATA:
      return "DATA";
    case STREAM_WIN32_DATA:
      return "WIN32-DATA";
    case STREAM_WIN32_GZIP_DATA:
      return "WIN32-GZIP";
    case STREAM_WIN32_COMPRESSED_DATA:
      return "WIN32-COMPRESSED";
    case STREAM_MD5_DIGEST:
      return "MD5";
    case STREAM_SHA1_DIGEST:
      return "SHA1";
    case STREAM_GZIP_DATA:
      return "GZIP";
    case STREAM_COMPRESSED_DATA:
      return "COMPRESSED";
    case STREAM_UNIX_ATTRIBUTES_EX:
      return "UNIX-Attributes-EX";
    case STREAM_RESTORE_OBJECT:
      return "RESTORE-OBJECT";
    case STREAM_SPARSE_DATA:
      return "SPARSE-DATA";
    case STREAM_SPARSE_GZIP_DATA:
      return "SPARSE-GZIP";
    case STREAM_SPARSE_COMPRESSED_DATA:
      return "SPARSE-COMPRESSED";
    case STREAM_PROGRAM_NAMES:
      return "PROG-NAMES";
    case STREAM_PROGRAM_DATA:
      return "PROG-DATA";
    case STREAM_PLUGIN_NAME:
      return "PLUGIN-NAME";
    case STREAM_MACOS_FORK_DATA:
      return "MACOS-RSRC";
    case STREAM_HFSPLUS_ATTRIBUTES:
      return "HFSPLUS-Attributes";
    case STREAM_SHA256_DIGEST:
      return "SHA256";
    case STREAM_SHA512_DIGEST:
      return "SHA512";
    case STREAM_SIGNED_DIGEST:
      return "SIGNED-DIGEST";
    case STREAM_ENCRYPTED_SESSION_DATA:
      return "ENCRYPTED-SESSION-DATA";
    case STREAM_ENCRYPTED_FILE_DATA:
      return "ENCRYPTED-FILE";
    case STREAM_ENCRYPTED_FILE_GZIP_DATA:
      return "ENCRYPTED-GZIP";
    case STREAM_ENCRYPTED_FILE_COMPRESSED_DATA:
      return "ENCRYPTED-COMPRESSED";
    case STREAM_ENCRYPTED_WIN32_DATA:
      return "ENCRYPTED-WIN32-DATA";
    case STREAM_ENCRYPTED_WIN32_GZIP_DATA:
      return "ENCRYPTED-WIN32-GZIP";
    case STREAM_ENCRYPTED_WIN32_COMPRESSED_DATA:
      return "ENCRYPTED-WIN32-COMPRESSED";
    case STREAM_ENCRYPTED_MACOS_FORK_DATA:
      return "ENCRYPTED-MACOS-RSRC";
    default:
      sprintf(buf, "%d", stream);
      return buf;
  }
}

static const char* record_state_to_ascii(rec_state state)
{
  switch (state) {
    case st_none:
      return "st_none";
    case st_header:
      return "st_header";
    case st_header_cont:
      return "st_header_cont";
    case st_data:
      return "st_data";
    default:
      return "<unknown>";
  }
}

static const char* findex_to_str(int32_t index, char* buf, size_t bufsz)
{
  if (index >= 0) {
    Bsnprintf(buf, bufsz, "<User> %d", index);
    return buf;
  }

  FI_to_ascii(buf, index);

  return buf;
}


static const char* record_unix_attributes_to_str(PoolMem& resultbuffer,
                                                 JobControlRecord* jcr,
                                                 const DeviceRecord* rec)
{
  Attributes* attr = new_attr(NULL);

  if (!UnpackAttributesRecord(jcr, rec->Stream, rec->data, rec->data_len,
                              attr)) {
    resultbuffer.bsprintf("ERROR");
    return NULL;
  }

  attr->data_stream =
      DecodeStat(attr->attr, &attr->statp, sizeof(attr->statp), &attr->LinkFI);
  BuildAttrOutputFnames(jcr, attr);
  attr_to_str(resultbuffer, jcr, attr);

  FreeAttr(attr);

  return resultbuffer.c_str();
}


static const char* get_record_short_info(PoolMem& resultbuffer,
                                         JobControlRecord* jcr,
                                         const DeviceRecord* rec)
{
  switch (rec->maskedStream) {
    case STREAM_UNIX_ATTRIBUTES:
    case STREAM_UNIX_ATTRIBUTES_EX:
      record_unix_attributes_to_str(resultbuffer, jcr, rec);
      break;
    case STREAM_MD5_DIGEST:
    case STREAM_SHA1_DIGEST:
    case STREAM_SHA256_DIGEST:
    case STREAM_SHA512_DIGEST:
      record_digest_to_str(resultbuffer, rec);
      break;
    case STREAM_PLUGIN_NAME: {
      char data[100];
      int len = MIN(rec->data_len + 1, sizeof(data));
      bstrncpy(data, rec->data, len);
      resultbuffer.bsprintf("data: %s\n", data);
      break;
    }
    case STREAM_RESTORE_OBJECT:
      resultbuffer.bsprintf("Restore Object record");
      break;
    case STREAM_COMPRESSED_DATA:
    case STREAM_SPARSE_COMPRESSED_DATA:
    case STREAM_WIN32_COMPRESSED_DATA:
    case STREAM_ENCRYPTED_FILE_COMPRESSED_DATA:
    case STREAM_ENCRYPTED_WIN32_COMPRESSED_DATA:
    case STREAM_GZIP_DATA:        /* deprecated */
    case STREAM_SPARSE_GZIP_DATA: /* deprecated */
      record_compression_to_str(resultbuffer, rec);
      break;
    default:
      break;
  }
  return resultbuffer.c_str();
}

const char* record_to_str(PoolMem& resultbuffer,
                          JobControlRecord* jcr,
                          const DeviceRecord* rec)
{
  PoolMem record_info_buf(PM_MESSAGE);
  char stream_buf[100];

  resultbuffer.bsprintf(
      "FileIndex=%-5d Stream=%-2d %-25s DataLen=%-5d", rec->FileIndex,
      rec->Stream, stream_to_ascii(stream_buf, rec->Stream, rec->FileIndex),
      rec->data_len);
  IndentMultilineString(
      resultbuffer, get_record_short_info(record_info_buf, jcr, rec), " | ");

  return resultbuffer.c_str();
}

void DumpRecord(const char* tag, const DeviceRecord* rec)
{
  char stream[128];
  char findex[128];

  Dmsg2(100, "%s: rec %p\n", tag, rec);

  Dmsg3(100, "%-14s next %p prev %p\n", "link", rec->link.next, rec->link.prev);
  Dmsg2(100, "%-14s %u\n", "File", rec->File);
  Dmsg2(100, "%-14s %u\n", "Block", rec->Block);
  Dmsg2(100, "%-14s %u\n", "VolSessionId", rec->VolSessionId);
  Dmsg2(100, "%-14s %u\n", "VolSessionTime", rec->VolSessionTime);
  Dmsg2(100, "%-14s %s\n", "FileIndex",
        findex_to_str(rec->FileIndex, findex, sizeof(findex)));
  Dmsg2(100, "%-14s %s\n", "Stream",
        stream_to_ascii(stream, rec->Stream, rec->FileIndex));
  Dmsg2(100, "%-14s %d\n", "maskedStream", rec->maskedStream);
  Dmsg2(100, "%-14s %u\n", "data_len", rec->data_len);
  Dmsg2(100, "%-14s %u\n", "remainder", rec->remainder);
  for (unsigned int i = 0;
       i < (sizeof(rec->state_bits) / sizeof(rec->state_bits[0])); i++) {
    Dmsg3(100, "%-11s[%d]        %2.2x\n", "state_bits", i,
          (uint8_t)rec->state_bits[i]);
  }
  Dmsg3(100, "%-14s %u (%s)\n", "state", rec->state,
        record_state_to_ascii(rec->state));
  Dmsg2(100, "%-14s %p\n", "bsr", rec->bsr);
  Dmsg2(100, "%-14s %p\n", "data", rec->data);
  Dmsg2(100, "%-14s %d\n", "match_stat", rec->match_stat);
  Dmsg2(100, "%-14s %u\n", "last_VolSessionId", rec->last_VolSessionId);
  Dmsg2(100, "%-14s %u\n", "last_VolSessionTime", rec->last_VolSessionTime);
  Dmsg2(100, "%-14s %d\n", "last_FileIndex", rec->last_FileIndex);
  Dmsg2(100, "%-14s %d\n", "last_Stream", rec->last_Stream);
  Dmsg2(100, "%-14s %s\n", "own_mempool", rec->own_mempool ? "true" : "false");
}

/**
 * Return a new record entity
 */
DeviceRecord* new_record(bool with_data)
{
  DeviceRecord* rec;

  rec = (DeviceRecord*)GetPoolMemory(PM_RECORD);
  *rec = DeviceRecord{};
  if (with_data) {
    rec->data = GetPoolMemory(PM_MESSAGE);
    rec->own_mempool = true;
  }
  rec->state = st_none;

  return rec;
}

void EmptyRecord(DeviceRecord* rec)
{
  rec->File = rec->Block = 0;
  rec->VolSessionId = rec->VolSessionTime = 0;
  rec->FileIndex = rec->Stream = 0;
  rec->data_len = rec->remainder = 0;

  ClearBit(REC_PARTIAL_RECORD, rec->state_bits);
  ClearBit(REC_BLOCK_EMPTY, rec->state_bits);
  ClearBit(REC_NO_MATCH, rec->state_bits);
  ClearBit(REC_CONTINUATION, rec->state_bits);

  rec->state = st_none;
}

void CopyRecordState(DeviceRecord* dst, DeviceRecord* src)
{
  bool own_mempool;
  int32_t Stream, maskedStream;
  uint32_t data_len;
  POOLMEM* data;

  /*
   * Preserve some important fields all other can be overwritten.
   */
  Stream = dst->Stream;
  maskedStream = dst->maskedStream;
  data = dst->data;
  data_len = dst->data_len;
  own_mempool = dst->own_mempool;

  memcpy(dst, src, sizeof(DeviceRecord));

  dst->Stream = Stream;
  dst->maskedStream = maskedStream;
  dst->data = data;
  dst->data_len = data_len;
  dst->own_mempool = own_mempool;
}

/**
 * Free the record entity
 */
void FreeRecord(DeviceRecord* rec)
{
  Dmsg0(950, "Enter FreeRecord.\n");
  if (rec->data && rec->own_mempool) { FreePoolMemory(rec->data); }
  Dmsg0(950, "Data buf is freed.\n");
  FreePoolMemory((POOLMEM*)rec);
  Dmsg0(950, "Leave FreeRecord.\n");
}

static inline ssize_t WriteHeaderToBlock(DeviceBlock* block,
                                         const DeviceRecord* rec,
                                         int32_t Stream)
{
  ser_declare;

  /*
   * Require enough room to write a full header
   */
  if (BlockWriteNavail(block) < WRITE_RECHDR_LENGTH) return -1;

  SerBegin(block->bufp, WRITE_RECHDR_LENGTH);

  if (BLOCK_VER == 1) {
    ser_uint32(rec->VolSessionId);
    ser_uint32(rec->VolSessionTime);
  } else {
    block->VolSessionId = rec->VolSessionId;
    block->VolSessionTime = rec->VolSessionTime;
  }

  ser_int32(rec->FileIndex);
  ser_int32(Stream);

  ser_uint32(
      rec->remainder); /* each header tracks remaining user bytes to write */

  block->bufp += WRITE_RECHDR_LENGTH;
  block->binbuf += WRITE_RECHDR_LENGTH;

  if (rec->FileIndex > 0) {
    /*
     * If data record, update what we have in this block
     */
    if (block->FirstIndex == 0) { block->FirstIndex = rec->FileIndex; }
    block->LastIndex = rec->FileIndex;
  }

  return WRITE_RECHDR_LENGTH;
}

static inline ssize_t WriteDataToBlock(DeviceBlock* block,
                                       const DeviceRecord* rec)
{
  uint32_t len;

  len = MIN(rec->remainder, BlockWriteNavail(block));
  memcpy(block->bufp,
         ((unsigned char*)rec->data) + (rec->data_len - rec->remainder), len);
  block->bufp += len;
  block->binbuf += len;
  return len;
}

/**
 * Write a Record to the block
 *
 * Returns: false means the block could not be written to tape/disk.
 *          true on success (all bytes written to the block).
 */
bool DeviceControlRecord::WriteRecord()
{
  bool retval = false;
  bool translated_record = false;
  char buf1[100], buf2[100];

  /*
   * Perform record translations.
   */
  before_rec = rec;
  after_rec = NULL;
  if (GeneratePluginEvent(jcr, bSdEventWriteRecordTranslation, this) !=
      bRC_OK) {
    goto bail_out;
  }

  /*
   * The record got translated when we got an after_rec pointer after calling
   * the bSdEventWriteRecordTranslation plugin event. If no translation has
   * taken place we just point the after_rec pointer to same DeviceRecord as in
   * the before_rec pointer.
   */
  if (!after_rec) {
    after_rec = before_rec;
  } else {
    translated_record = true;
  }

  while (!WriteRecordToBlock(this, after_rec)) {
    Dmsg2(850, "!WriteRecordToBlock data_len=%d rem=%d\n", after_rec->data_len,
          after_rec->remainder);
    if (!WriteBlockToDevice()) {
      Dmsg2(90, "Got WriteBlockToDev error on device %s. %s\n",
            dev->print_name(), dev->bstrerror());
      goto bail_out;
    }
  }

  jcr->JobBytes += after_rec->data_len; /* increment bytes this job */
  if (jcr->impl->RemainingQuota && jcr->JobBytes > jcr->impl->RemainingQuota) {
    Jmsg0(jcr, M_FATAL, 0, _("Quota Exceeded. Job Terminated.\n"));
    goto bail_out;
  }

  Dmsg4(850, "WriteRecord FI=%s SessId=%d Strm=%s len=%d\n",
        FI_to_ascii(buf1, after_rec->FileIndex), after_rec->VolSessionId,
        stream_to_ascii(buf2, after_rec->Stream, after_rec->FileIndex),
        after_rec->data_len);

  retval = true;

bail_out:
  if (translated_record) {
    CopyRecordState(before_rec, after_rec);
    FreeRecord(after_rec);
    after_rec = NULL;
  }

  return retval;
}

/**
 * Write a Record to the block
 *
 *  Returns: false on failure (none or partially written)
 *           true  on success (all bytes written)
 *
 *  and remainder returned in packet.
 *
 *  We require enough room for the header, and we deal with
 *  two special cases. 1. Only part of the record may have
 *  been transferred the last time (when remainder is
 *  non-zero), and 2. The remaining bytes to write may not
 *  all fit into the block.
 */
bool WriteRecordToBlock(DeviceControlRecord* dcr, DeviceRecord* rec)
{
  ssize_t n;
  bool retval = false;
  char buf1[100], buf2[100];
  DeviceBlock* block = dcr->block;

  /*
   * After this point the record is in nrec not rec e.g. its either converted
   * or is just a pointer to the same as the rec pointer being passed in.
   */

  while (1) {
    ASSERT(block->binbuf == (uint32_t)(block->bufp - block->buf));
    ASSERT(block->buf_len >= block->binbuf);

    Dmsg9(890,
          "%s() state=%d (%s) FI=%s SessId=%d Strm=%s len=%d "
          "block_navail=%d remainder=%d\n",
          __PRETTY_FUNCTION__, rec->state, record_state_to_ascii(rec->state),
          FI_to_ascii(buf1, rec->FileIndex), rec->VolSessionId,
          stream_to_ascii(buf2, rec->Stream, rec->FileIndex), rec->data_len,
          BlockWriteNavail(block), rec->remainder);

    switch (rec->state) {
      case st_none:
        /*
         * Figure out what to do
         */
        rec->state = st_header;
        rec->remainder = rec->data_len; /* length of data remaining to write */
        continue;                       /* goto st_header */

      case st_header:
        /*
         * Write header
         */
        n = WriteHeaderToBlock(block, rec, rec->Stream);
        if (n < 0) {
          /*
           * the header did not fit into the block, so flush the current
           * block and come back to st_header and try again on the next block.
           */
          goto bail_out;
        }

        if (BlockWriteNavail(block) == 0) {
          /*
           * The header fit, but no bytes of data will fit,
           * so flush this block and start the next block with a
           * continuation header.
           */
          rec->state = st_header_cont;
          goto bail_out;
        }

        /*
         * The header fit, and at least one byte of data will fit,
         * so move to the st_data state and start filling the block
         * with data bytes
         */
        rec->state = st_data;
        continue;

      case st_header_cont:
        /*
         * Write continuation header
         */
        n = WriteHeaderToBlock(block, rec, -rec->Stream);
        if (n < 0) {
          /*
           * The continuation header wouldn't fit, which is impossible
           * unless something is broken
           */
          Emsg0(M_ABORT, 0, _("couldn't write continuation header\n"));
        }

        /*
         * After successfully writing a continuation header, we always start
         * writing data, even if none will fit into this block.
         */
        rec->state = st_data;

        if (BlockWriteNavail(block) == 0) {
          /*
           * The header fit, but no bytes of data will fit,
           * so flush the block and start the next block with
           * data bytes
           */
          goto bail_out; /* Partial transfer */
        }

        continue;

      case st_data:
        /*
         * Write normal data
         *
         * Part of it may have already been transferred, and we
         * may not have enough room to transfer the whole this time.
         */
        if (rec->remainder > 0) {
          n = WriteDataToBlock(block, rec);
          if (n < 0) {
            /*
             * error appending data to block should be impossible
             * unless something is broken
             */
            Emsg0(M_ABORT, 0, _("data write error\n"));
          }

          rec->remainder -= n;

          if (rec->remainder > 0) {
            /*
             * Could not fit all of the data bytes into this block, so
             * flush the current block, and start the next block with a
             * continuation header
             */
            rec->state = st_header_cont;
            goto bail_out;
          }
        }

        rec->remainder = 0; /* did whole transfer */
        rec->state = st_none;
        retval = true;
        goto bail_out;

      default:
        Emsg1(M_ABORT, 0, _("Something went wrong. Unknown state %d.\n"),
              rec->state);
        rec->state = st_none;
        retval = true;
        goto bail_out;
    }
  }

bail_out:
  return retval;
}

/**
 * Test if we can write whole record to the block
 *
 *  Returns: false on failure
 *           true  on success (all bytes can be written)
 */
bool CanWriteRecordToBlock(DeviceBlock* block, const DeviceRecord* rec)
{
  return BlockWriteNavail(block) >= WRITE_RECHDR_LENGTH + rec->remainder;
}

uint64_t GetRecordAddress(const DeviceRecord* rec)
{
  return ((uint64_t)rec->File) << 32 | rec->Block;
}

/**
 * Read a Record from the block
 *
 * Returns: false if nothing read or if the continuation record does not match.
 *                In both of these cases, a block read must be done.
 *          true  if at least the record header was read, this
 *                routine may have to be called again with a new
 *                block if the entire record was not read.
 */
bool ReadRecordFromBlock(DeviceControlRecord* dcr, DeviceRecord* rec)
{
  ser_declare;
  uint32_t remlen;
  uint32_t VolSessionId;
  uint32_t VolSessionTime;
  int32_t FileIndex;
  int32_t Stream;
  uint32_t data_bytes;
  uint32_t rhl;
  char buf1[100], buf2[100];

  remlen = dcr->block->binbuf;

  /*
   * Clear state flags
   */
  ClearAllBits(REC_STATE_MAX, rec->state_bits);
  if (dcr->block->dev->IsTape()) { SetBit(REC_ISTAPE, rec->state_bits); }
  rec->Block = ((Device*)(dcr->block->dev))->EndBlock;
  rec->File = ((Device*)(dcr->block->dev))->EndFile;

  /*
   * Get the header. There is always a full header, otherwise we find it in the
   * next block.
   */
  Dmsg3(450, "Block=%d Ver=%d size=%u\n", dcr->block->BlockNumber,
        dcr->block->BlockVer, dcr->block->block_len);
  if (dcr->block->BlockVer == 1) {
    rhl = RECHDR1_LENGTH;
  } else {
    rhl = RECHDR2_LENGTH;
  }
  if (remlen >= rhl) {
    Dmsg4(450,
          "Enter read_record_block: remlen=%d data_len=%d rem=%d blkver=%d\n",
          remlen, rec->data_len, rec->remainder, dcr->block->BlockVer);

    UnserBegin(dcr->block->bufp, WRITE_RECHDR_LENGTH);
    if (dcr->block->BlockVer == 1) {
      unser_uint32(VolSessionId);
      unser_uint32(VolSessionTime);
    } else {
      VolSessionId = dcr->block->VolSessionId;
      VolSessionTime = dcr->block->VolSessionTime;
    }
    unser_int32(FileIndex);
    unser_int32(Stream);
    unser_uint32(data_bytes);

    dcr->block->bufp += rhl;
    dcr->block->binbuf -= rhl;
    remlen -= rhl;

    /*
     * If we are looking for more (remainder!=0), we reject anything
     * where the VolSessionId and VolSessionTime don't agree
     */
    if (rec->remainder && (rec->VolSessionId != VolSessionId ||
                           rec->VolSessionTime != VolSessionTime)) {
      SetBit(REC_NO_MATCH, rec->state_bits);
      Dmsg0(450, "remainder and VolSession doesn't match\n");
      return false; /* This is from some other Session */
    }

    /*
     * If Stream is negative, it means that this is a continuation
     * of a previous partially written record.
     */
    if (Stream < 0) { /* continuation record? */
      Dmsg1(500, "Got negative Stream => continuation. remainder=%d\n",
            rec->remainder);
      SetBit(REC_CONTINUATION, rec->state_bits);
      if (!rec->remainder) { /* if we didn't read previously */
        rec->data_len = 0;   /* return data as if no continuation */
      } else if (rec->Stream != -Stream) {
        SetBit(REC_NO_MATCH, rec->state_bits);
        return false; /* This is from some other Session */
      }
      rec->Stream = -Stream; /* set correct Stream */
      rec->maskedStream = rec->Stream & STREAMMASK_TYPE;
    } else { /* Regular record */
      rec->Stream = Stream;
      rec->maskedStream = rec->Stream & STREAMMASK_TYPE;
      rec->data_len = 0; /* transfer to beginning of data */
    }
    rec->VolSessionId = VolSessionId;
    rec->VolSessionTime = VolSessionTime;
    rec->FileIndex = FileIndex;
    if (FileIndex > 0) {
      if (dcr->block->FirstIndex == 0) { dcr->block->FirstIndex = FileIndex; }
      dcr->block->LastIndex = FileIndex;
    }

    Dmsg6(450,
          "rd_rec_blk() got FI=%s SessId=%d Strm=%s len=%u\n"
          "remlen=%d data_len=%d\n",
          FI_to_ascii(buf1, rec->FileIndex), rec->VolSessionId,
          stream_to_ascii(buf2, rec->Stream, rec->FileIndex), data_bytes,
          remlen, rec->data_len);
  } else {
    /*
     * No more records in this block because the number
     * of remaining bytes are less than a record header
     * length, so return empty handed, but indicate that
     * he must read again. By returning, we allow the
     * higher level routine to fetch the next block and
     * then reread.
     */
    Dmsg0(450, "read_record_block: nothing\n");
    SetBit(REC_NO_HEADER, rec->state_bits);
    SetBit(REC_BLOCK_EMPTY, rec->state_bits);
    EmptyBlock(dcr->block); /* mark block empty */
    return false;
  }

  /* Sanity check */
  if (data_bytes >= MAX_BLOCK_LENGTH) {
    /*
     * Something is wrong, force read of next block, abort
     *   continuing with this block.
     */
    SetBit(REC_NO_HEADER, rec->state_bits);
    SetBit(REC_BLOCK_EMPTY, rec->state_bits);
    EmptyBlock(dcr->block);
    Jmsg2(dcr->jcr, M_WARNING, 0,
          _("Sanity check failed. maxlen=%d datalen=%d. Block discarded.\n"),
          MAX_BLOCK_LENGTH, data_bytes);
    return false;
  }

  rec->data = CheckPoolMemorySize(rec->data, rec->data_len + data_bytes);

  /*
   * At this point, we have read the header, now we
   * must transfer as much of the data record as
   * possible taking into account: 1. A partial
   * data record may have previously been transferred,
   * 2. The current block may not contain the whole data
   * record.
   */
  if (remlen >= data_bytes) {
    /*
     * Got whole record
     */
    memcpy(rec->data + rec->data_len, dcr->block->bufp, data_bytes);
    dcr->block->bufp += data_bytes;
    dcr->block->binbuf -= data_bytes;
    rec->data_len += data_bytes;
  } else {
    /*
     * Partial record
     */
    memcpy(rec->data + rec->data_len, dcr->block->bufp, remlen);
    dcr->block->bufp += remlen;
    dcr->block->binbuf -= remlen;
    rec->data_len += remlen;
    rec->remainder = 1; /* partial record transferred */
    Dmsg1(450, "read_record_block: partial xfered=%d\n", rec->data_len);
    SetBit(REC_PARTIAL_RECORD, rec->state_bits);
    SetBit(REC_BLOCK_EMPTY, rec->state_bits);
    return true;
  }
  rec->remainder = 0;

  Dmsg4(450, "Rtn full rd_rec_blk FI=%s SessId=%d Strm=%s len=%d\n",
        FI_to_ascii(buf1, rec->FileIndex), rec->VolSessionId,
        stream_to_ascii(buf2, rec->Stream, rec->FileIndex), rec->data_len);

  return true; /* transferred full record */
}

} /* namespace storagedaemon */
